package cn.itcast.spark.sql

import org.apache.spark.sql.SparkSession
import org.junit.Test

class UntypedTransformation {
  val spark = SparkSession.builder()
    .master("local[6]")
    .appName("untyped")
    .getOrCreate()

  import spark.implicits._

  @Test
  def select(): Unit ={
    val ds = Seq(Person("zhangsan", 12), Person("lisi", 18), Person("zhangsan", 8)).toDS()
    // 在dataset中，select可以在任何位置调用
    ds.select('name).show()
    // 给出表达式
    ds.selectExpr("sum(age)").show()

    import org.apache.spark.sql.functions._

    ds.select(expr("sum(age)")).show()
  }

  @Test
  def column(): Unit = {
    val ds = Seq(Person("zhangsan", 12), Person("lisi", 18), Person("zhangsan", 8)).toDS()

    import org.apache.spark.sql.functions._

    // select rand() from ...
    // 使用函数的方法
    //  1. 使用functions.xx
    //  2. 使用expr()表达式
    ds.withColumn("random", expr("rand()")).show()

    ds.withColumn("name_new", 'name).show()

    ds.withColumn("name_jok", 'name === "").show()

    ds.withColumnRenamed("name", "renamed").show()
  }

  @Test
  def groupBy(): Unit = {
    val ds = Seq(Person("zhangsan", 12), Person("zhangsan", 8), Person("lisi", 15)).toDS()

    // 为什么groupByKey是有类型的？因为groupByKey所生成的对象中的算子是有类型的
//    ds.groupByKey(item => item.name).mapValues()

    // 为什么groupBy是无类型的？因为groupBy所生成的对象中的算子是无类型的，针对列进行处理
    import org.apache.spark.sql.functions._
    ds.groupBy('name).agg(mean("age")).show()
  }

}
